package org.future.task.core.group;

import org.future.task.core.handler.Connection;
import org.future.task.core.util.ThreadPoolUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.*;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.stream.Collectors;

public abstract class AbstractGroup implements Group {

    Logger logger = LoggerFactory.getLogger(this.getClass());

    protected int commitThread;

    protected String sinkQueue;

    protected Connection receiveConnection;

    protected Connection sinkConnection;

    private ThreadPoolExecutor executorService;

    private final int batchSize;

    protected Set<String> queueSet;

    protected String splitStr;

    protected String queueStr;


    public AbstractGroup(int commitThread, String sinkQueue, Connection receiveConnection, Connection sinkConnection, int batchSize, Set<String> queueSet) {
        this.commitThread = commitThread;
        this.sinkQueue = sinkQueue;
        this.receiveConnection = receiveConnection;
        this.sinkConnection = sinkConnection;
        this.batchSize = batchSize;
        this.queueSet = queueSet == null ? new HashSet<>() : queueSet;
        init();
    }

    public AbstractGroup(int commitThread, String sinkQueue, Connection receiveConnection, Connection sinkConnection, int batchSize, String splitStr, String queueStr) {
        this.commitThread = commitThread;
        this.sinkQueue = sinkQueue;
        this.receiveConnection = receiveConnection;
        this.sinkConnection = sinkConnection;
        this.batchSize = batchSize;
        this.queueSet = (queueStr != null && splitStr != null && !"".equals(queueStr) && !"".equals(splitStr)) ? Arrays.stream(queueStr.split(splitStr)).collect(Collectors.toSet()) : new HashSet<>();//适用于JDK>=1.8
        init();
    }

    public AbstractGroup(int commitThread, String sinkQueue, Connection receiveConnection, Connection sinkConnection, int batchSize) {
        this(commitThread, sinkQueue, receiveConnection, sinkConnection, batchSize, null);
    }

    public AbstractGroup(int commitThread, String sinkQueue, Connection receiveConnection, Connection sinkConnection) {
        this(commitThread, sinkQueue, receiveConnection, sinkConnection, 0);
    }

    @Override
    public void submit(String queueName) {
        createThreadPool();
        synchronized (executorService) {
            executorService.setCorePoolSize(executorService.getCorePoolSize() + 1);
        }
        executorService.submit(new PollThread(queueName));
        logger.info("消费线程{}加入线程池", queueName);
    }

    void createThreadPool() {
        if (executorService == null) {
            synchronized (this) {
                if (executorService == null) {
                    this.executorService = new ThreadPoolUtils().getDefaultThreadPool("group", 1);
                }
            }
        }
    }

    @Override
    public void init() {
        createThreadPool();
        //提交线程
        for (String queueName : queueSet) {
            this.submit(queueName);
        }
    }

    class PollThread implements Runnable {

        private final String queueName;

        public PollThread(String queueName) {
            this.queueName = queueName;
        }

        @Override
        public void run() {
            while (true) {
                try {
                    List<Object> dataList = receive(queueName, batchSize <= 0 ? 100 : batchSize, 100);
                    if (dataList == null || dataList.isEmpty()) {
                        Thread.sleep(1);
                        continue;
                    }
                    sink(sinkQueue, dataList);
                } catch (Exception e) {
                    e.printStackTrace();
                }

            }
        }
    }

}
